package file_store

import (
	"bufio"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/hashicorp/raft"
	raftboltdb "github.com/hashicorp/raft-boltdb/v2"
	"io"
	"io/fs"
	"log"
	"net"
	"os"
	"path/filepath"
	"raftkv/utils"
	"strconv"
	"sync"
	"time"
)

var (
	// ErrOpenTimeout is returned when the Store does not apply its initial
	// logs within the specified time.
	ErrOpenTimeout = errors.New("timeout waiting for initial logs application")
)

const (
	retainSnapshotCount = 2
	raftTimeout         = 10 * time.Second
	leaderWaitDelay     = 100 * time.Millisecond
	appliedWaitDelay    = 100 * time.Millisecond
)

//指令
type command struct {
	Op       string `json:"op,omitempty"`
	Hash     string `json:"hash,omitempty"` //分片hash，用作区分文件
	CurIndex int    `json:"cur_index"`      //当前分片号
	Total    int    `json:"total,omitempty"`
	Key   string `json:"key"`
	Value string `json:"value"`
	Chunk	 []byte
}

type Store struct {
	lock     sync.Mutex
	m        map[string]string // store node info.
	Data     *[]byte           `json:"value,omitempty"` //file Data
	raft     *raft.Raft        // The consensus mechanism
	logger   *log.Logger
	RaftDir  string
	RaftBind string
}

func NewFileStore() *Store {
	return &Store{
		logger: log.New(os.Stderr, "[store] ", log.LstdFlags),
	}
}

func (s *Store) LeaderAddr() string {
	return string(s.raft.Leader())
}

func (s *Store) LeaderID() (string, error) {
	addr := s.LeaderAddr()
	configFuture := s.raft.GetConfiguration()
	if err := configFuture.Error(); err != nil {
		s.logger.Printf("failed to get raft configuration: %v", err)
		return "", err
	}

	for _, srv := range configFuture.Configuration().Servers {
		if srv.Address == raft.ServerAddress(addr) {
			return string(srv.ID), nil
		}
	}
	return "", nil
}
func (s *Store) LeaderAPIAddr() string {
	id, err := s.LeaderID()
	if err != nil {
		return ""
	}
	addr, err := s.GetMeta(id)
	if err != nil {
		return ""
	}
	return addr
}
func (s *Store) SetMeta(key, value string) error {
	return s.Set(key, value)
}
func (s *Store) Set(key, value string) error {
	if s.raft.State() != raft.Leader {
		return raft.ErrNotLeader
	}

	c := &command{
		Op:    "set",
		Key:   key,
		Value: value,
	}
	b, err := json.Marshal(c)
	if err != nil {
		return err
	}
	f := s.raft.Apply(b, raftTimeout)
	return f.Error()
}


// WaitForLeader wait for leader boom boom.
func (s *Store) WaitForLeader(timeout time.Duration) (string, error) {
	tck := time.NewTicker(leaderWaitDelay)
	defer tck.Stop()
	tmr := time.NewTimer(timeout)
	defer tmr.Stop()

	for {
		select {
		case <-tck.C:
			l := s.LeaderAddr()
			if l != "" {
				return l, nil
			}
		case <-tmr.C:
			return "", fmt.Errorf("timeout expired")
		}
	}
}

func (s *Store) WaitForAppliedIndex(idx uint64, timeout time.Duration) error {
	tck := time.NewTicker(appliedWaitDelay)
	defer tck.Stop()
	tmr := time.NewTimer(timeout)
	defer tmr.Stop()

	for {
		select {
		case <-tck.C:
			if s.raft.AppliedIndex() >= idx {
				return nil
			}
		case <-tmr.C:
			return fmt.Errorf("timeout expired")
		}
	}
}
func (s *Store) WaitForApplied(timeout time.Duration) error {
	if timeout == 0 {
		return nil
	}
	s.logger.Printf("waiting for up to %s for application of initial logs", timeout)
	if err := s.WaitForAppliedIndex(s.raft.LastIndex(), timeout); err != nil {
		return ErrOpenTimeout
	}
	return nil
}

func (s *Store) GetMeta(key string) (string, error) {
	return s.Get(key)
}

func (s *Store) Get(key string) (string, error) {
	s.lock.Lock()
	defer s.lock.Unlock()
	return s.m[key], nil
}
func (s *Store) StoreFile(hash string, total int, curIndex int, chunk []byte) error {
	if s.raft.State() != raft.Leader {
		return raft.ErrNotLeader
	}
	c := &command{
		Op:       "store",
		Hash:     hash,
		Total:    total,
		CurIndex: curIndex,
		Chunk: chunk,
	}
	data, err := json.Marshal(c)
	if err != nil {
		return err
	}
	f := s.raft.Apply(data, raftTimeout)
	return f.Error()
}

// Open 启动类 TODO:my code
func (s *Store) Open(enableSingle bool, localID string) error {
	config := raft.DefaultConfig()
	config.LocalID = raft.ServerID(localID)

	newNode := !utils.PathExists(filepath.Join(s.RaftDir, "raft.db"))
	addr, err := net.ResolveTCPAddr("tcp", s.RaftBind)
	if err != nil {
		return err
	}
	transport, err := raft.NewTCPTransport(s.RaftBind, addr, 3, 10*time.Second, os.Stderr)
	if err != nil {
		return err
	}

	snapshots, err := raft.NewFileSnapshotStore(s.RaftDir, retainSnapshotCount, os.Stderr)
	if err != nil {
		return fmt.Errorf("file snapshot store: %s", err)
	}

	var logStore raft.LogStore
	var stableStore raft.StableStore

	boltDB, err := raftboltdb.NewBoltStore(filepath.Join(s.RaftDir, "raft.db"))
	if err != nil {
		return fmt.Errorf("new bolt store: %s", err)
	}
	logStore = boltDB
	stableStore = boltDB
	ra, err := raft.NewRaft(config, (*fsm)(s), logStore, stableStore, snapshots, transport)
	if err != nil {
		return fmt.Errorf("new raft: %s", err)
	}
	s.raft = ra

	if enableSingle && newNode {
		s.logger.Printf("bootstrap needed")
		configuration := raft.Configuration{
			Servers: []raft.Server{
				{
					ID:      config.LocalID,
					Address: transport.LocalAddr(),
				},
			},
		}
		ra.BootstrapCluster(configuration)
	} else {
		s.logger.Printf("no bootstrap needed")
	}

	return nil
}
type fsm Store

// Apply applies a Raft log entry to the key-value store.
func (f *fsm) Apply(l *raft.Log) interface{} {
	var c command
	if err := json.Unmarshal(l.Data, &c); err != nil {
		panic(fmt.Sprintf("failed to unmarshal command: %s", err.Error()))
	}

	switch c.Op {
	case "set":
		return f.applySet(c.Key,c.Value)
	case "store":
		return f.applyStore(c.Hash,c.CurIndex,c.Total,c.Chunk)
	case "merge":
		return f.applyMerge(c.Hash)
	default:
		panic(fmt.Sprintf("unrecognized command op: %s", c.Op))
	}
}

// Snapshot returns a snapshot of the key-value store.
func (f *fsm) Snapshot() (raft.FSMSnapshot, error) {
	f.lock.Lock()
	defer f.lock.Unlock()

	// Clone the map.
	o := make(map[string]string)
	for k, v := range f.m {
		o[k] = v
	}
	return &fsmSnapshot{store: o}, nil
}

// Restore stores the key-value store to a previous state.
func (f *fsm) Restore(rc io.ReadCloser) error {
	o := make(map[string]string)
	if err := json.NewDecoder(rc).Decode(&o); err != nil {
		return err
	}

	// Set the state from the snapshot, no lock required according to
	// Hashicorp docs.
	f.m = o
	return nil
}

func (f *fsm) applySet(key, value string) interface{} {
	f.lock.Lock()
	defer f.lock.Unlock()
	f.m[key] = value
	return nil
}

func (f *fsm) applyDelete(key string) interface{} {
	f.lock.Lock()
	defer f.lock.Unlock()
	delete(f.m, key)
	return nil
}

func (f *fsm) applyStore(hash string, index int, total int, chunk []byte) interface{} {
	f.lock.Lock()
	defer f.lock.Unlock()
	key := hash + strconv.Itoa(index)
	f.m[key]="1"
	_, err := os.Stat("./" + hash)
	if err!=nil{
		err = os.MkdirAll("./" + hash, os.ModeAppend)
		if err != nil{
			f.logger.Printf("error create dir:{%v}\n",hash)
		}
	}
	filename := "./" + strconv.Itoa(index) + ".chunk"
	err = os.WriteFile(filename, chunk, fs.ModeAppend)
	if err !=nil{
		f.logger.Printf("create file error , fn is {%v}\n", filename)
		return err
	}
	return nil
}
//处理合并
func (f *fsm) applyMerge(hash string) interface{} {
	f.lock.Lock()
	defer f.lock.Unlock()
	dirPath := "./" + hash
	_, err := os.Stat(dirPath)
	if err !=nil{
		f.logger.Printf("merge error\n")
	}
	dirs, err := os.ReadDir(dirPath)
	mergePath := "./fin/"
	file, err := os.Create(mergePath + hash+".png")
	defer file.Close()
	writer := bufio.NewWriter(file)
	for _,file := range dirs{
		filecontent, err := os.ReadFile(dirPath + "/" + file.Name())
		if err !=nil{
			fmt.Println(err)
		}
		_, err = writer.Write(filecontent)
		if err != nil {
			return nil
		}
	}
	writer.Flush()
	return nil
}

type fsmSnapshot struct {
	store map[string]string
}
func (s *Store) Join(nodeID, httpAddr string, addr string) error {
	s.logger.Printf("received join request for remote node %s at %s\n", nodeID, addr)
	configFuture := s.raft.GetConfiguration()
	if err := configFuture.Error(); err != nil {
		s.logger.Printf("failed to get raft configuration: %v", err)
		return err
	}
	for _, srv := range configFuture.Configuration().Servers {
		if srv.ID == raft.ServerID(nodeID) || srv.Address == raft.ServerAddress(addr) {
			if srv.Address == raft.ServerAddress(addr) && srv.ID == raft.ServerID(nodeID) {
				s.logger.Printf("node %s at %s already member of cluster, ignoring join request", nodeID, addr)
				return nil
			}

			future := s.raft.RemoveServer(srv.ID, 0, 0)
			if err := future.Error(); err != nil {
				return fmt.Errorf("error removing existing node %s at %s: %s", nodeID, addr, err)
			}
		}
	}

	f := s.raft.AddVoter(raft.ServerID(nodeID), raft.ServerAddress(addr), 0, 0)
	if f.Error() != nil {
		return f.Error()
	}

	if err := s.SetMeta(nodeID, httpAddr); err != nil {
		return err
	}

	s.logger.Printf("node %s at %s joined successfully", nodeID, addr)
	return nil
}
func (f *fsmSnapshot) Persist(sink raft.SnapshotSink) error {
	err := func() error {
		// Encode data.
		b, err := json.Marshal(f.store)
		if err != nil {
			return err
		}

		// Write data to sink.
		if _, err := sink.Write(b); err != nil {
			return err
		}

		// Close the sink.
		return sink.Close()
	}()

	if err != nil {
		sink.Cancel()
		return err
	}

	return nil
}

func (f *fsmSnapshot) Release() {}
